package com.btioe.yugong.producer;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry.*;
import com.alibaba.otter.canal.protocol.Message;
import com.btioe.yugong.common.conf.ConfigLoader;
import com.btioe.yugong.common.conf.ProfileBO;
import com.btioe.yugong.comsumer.SqlExcuter;
import com.google.protobuf.InvalidProtocolBufferException;
import com.iote.rock.common.exception.BizException;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

import static com.btioe.yugong.common.exe.ErrorCodeEnum.PROFILE_IS_NULL;

@Slf4j
public class CanalClient {
    private static final String DATA_VERSION_FLAG_FIELD_NAME = "version";
    private static final String VM_PROFILE_NAME = "profile";
    private static final String MysqlType_BIGINT = "bigint";
    private static final String MysqlType_VARCHAR = "varchar";
    private static final String MysqlType_DATETIME = "datetime";
    private static final String MysqlType_TINYINT = "tinyint";
    /**
     * 执行线程池
     * 33kafka消费子线程、1kafka消费父线程、1sms消息消费线程、1动态二维码信息传递线程
     */
    private static ExecutorService executorService = Executors.newFixedThreadPool(13);
    private static LinkedBlockingQueue[] queues;

    public static void main(String args[]) {
        String profile = System.getProperty(VM_PROFILE_NAME);
        if (StringUtils.isEmpty(profile)) {
            throw BizException.create(PROFILE_IS_NULL);
        }
        log.info("profile:{}", profile);
        ProfileBO profileBO = ConfigLoader.getProfile(profile);

        queues = new LinkedBlockingQueue[10];
        for (int i = 0; i < 10; i++) {
            queues[i] = new LinkedBlockingQueue();
        }
        executorService.execute(new PullBinLogTask(profileBO.getCanalUrl(), profileBO.getCanalPort()));
        for (int index = 0; index < 10; index++) {
            executorService.execute(new SqlExcuter(profileBO.getCDbUrl(),
                    profileBO.getCDbName(),
                    profileBO.getCDbUser(),
                    profileBO.getCDbPwd(),
                    queues[index]));
        }
    }

    static class PullBinLogTask implements Runnable {
        String hostname;
        Integer port;

        PullBinLogTask(String hostname, Integer port) {
            this.hostname = hostname;
            this.port = port;
        }

        @Override
        public void run() {

            CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(hostname, port), "example", "", "");
            int batchSize = 1000;
            try {
                connector.connect();
                connector.subscribe("");
                connector.rollback();
                while (true) {
                    try {
                        //尝试从master那边拉去数据batchSize条记录，有多少取多少
                        Message message = connector.getWithoutAck(batchSize);
                        long batchId = message.getId();
                        int size = message.getEntries().size();
                        if (batchId == -1 || size == 0) {
                            Thread.sleep(1000);
                        } else {
                            dataHandle(message.getEntries());
                        }
                        connector.ack(batchId);
                    } catch (Exception e) {
                        log.error("", e);
                    }
                }

            } finally {
                connector.disconnect();
            }
        }
    }

    /**
     * 数据处理
     *
     * @param entrys
     */
    private static void dataHandle(List<Entry> entrys) throws InvalidProtocolBufferException {
        for (Entry entry : entrys) {
            if (EntryType.ROWDATA == entry.getEntryType()) {
                RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());
                EventType eventType = rowChange.getEventType();
                if (eventType == EventType.DELETE) {
                    saveDeleteSql(entry);
                } else if (eventType == EventType.UPDATE) {
                    saveUpdateSql(entry);
                } else if (eventType == EventType.INSERT) {
                    saveInsertSql(entry);
                }
            }
        }
    }

    /**
     * 保存更新语句
     *
     * @param entry
     */
    private static void saveUpdateSql(Entry entry) {
        try {
            RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());
            List<RowData> rowDatasList = rowChange.getRowDatasList();
            for (RowData rowData : rowDatasList) {
                Column versionColum = null;
                List<Column> newColumnList = rowData.getAfterColumnsList();
                StringBuffer sql = new StringBuffer("update " + entry.getHeader().getSchemaName() + "." + entry.getHeader().getTableName() + " set ");
                for (int i = 0; i < newColumnList.size(); i++) {
                    Column column = newColumnList.get(i);
                    if (DATA_VERSION_FLAG_FIELD_NAME.equals(column.getName())) {
                        versionColum = column;
                    }
                    sql.append(" " + column.getName()
                            + " = " + buidlVle(column));
                    if (i != newColumnList.size() - 1) {
                        sql.append(",");
                    }
                }
                sql.append(" where ");
                List<Column> oldColumnList = rowData.getBeforeColumnsList();
                Column keyClm = null;
                for (Column column : oldColumnList) {
                    if (column.getIsKey()) {
                        keyClm = column;
                        //暂时只支持单一主键
                        sql.append(column.getName() + "=" + column.getValue());
                        break;
                    }
                }

                if (!Objects.isNull(versionColum)) {
                    sql.append(" and ").append(versionColum.getName() + "<" + versionColum.getValue());
                }

                int i = (keyClm.getValue().hashCode() & 0x7FFFFFFF) % 10;
                queues[i].add(sql.toString());
            }
        } catch (InvalidProtocolBufferException e) {
            e.printStackTrace();
        }
    }

    /**
     * 保存删除语句
     *
     * @param entry
     */
    private static void saveDeleteSql(Entry entry) {
        try {
            RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());
            List<RowData> rowDatasList = rowChange.getRowDatasList();
            for (RowData rowData : rowDatasList) {
                List<Column> columnList = rowData.getBeforeColumnsList();
                StringBuffer sql = new StringBuffer("delete from " + entry.getHeader().getSchemaName() + "." + entry.getHeader().getTableName() + " where ");
                for (Column column : columnList) {
                    if (column.getIsKey()) {
                        //暂时只支持单一主键
                        sql.append(column.getName() + "=" + column.getValue());
                        break;
                    }
                }
                Column keyClm = null;
                for (Column column : columnList) {
                    if (column.getIsKey()) {
                        keyClm = column;
                        break;
                    }
                }
                int i = (keyClm.getValue().hashCode() & 0x7FFFFFFF) % 10;
                queues[i].add(sql.toString());
            }
        } catch (InvalidProtocolBufferException e) {
            e.printStackTrace();
        }
    }

    /**
     * 保存插入语句
     *
     * @param entry
     */
    private static void saveInsertSql(Entry entry) {
        try {
            RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());
            List<RowData> rowDatasList = rowChange.getRowDatasList();
            for (RowData rowData : rowDatasList) {
                List<Column> columnList = rowData.getAfterColumnsList();
                StringBuffer sql = new StringBuffer("insert into " + entry.getHeader().getSchemaName() + "." + entry.getHeader().getTableName() + " (");
                for (int i = 0; i < columnList.size(); i++) {
                    sql.append(columnList.get(i).getName());
                    if (i != columnList.size() - 1) {
                        sql.append(",");
                    }
                }
                sql.append(") VALUES (");
                for (int i = 0; i < columnList.size(); i++) {
                    Column column = columnList.get(i);
                    sql.append(buidlVle(column));
                    if (i != columnList.size() - 1) {
                        sql.append(",");
                    }
                }
                sql.append(")");
                Column keyClm = null;
                for (Column column : columnList) {
                    if (column.getIsKey()) {
                        keyClm = column;
                        break;
                    }
                }
                int i = (keyClm.getValue().hashCode() & 0x7FFFFFFF) % 10;
                queues[i].add(sql.toString());
            }
        } catch (InvalidProtocolBufferException e) {
            e.printStackTrace();
        }
    }

    private static Object buidlVle(Column column) {
        String mysqlType = column.getMysqlType();
        if (mysqlType.startsWith(MysqlType_BIGINT)) {
            return StringUtils.isBlank(column.getValue()) ? 0 : column.getValue();
        } else if (mysqlType.startsWith(MysqlType_DATETIME)) {
            return StringUtils.isBlank(column.getValue()) ? null : "'" + column.getValue() + "'";
        } else if (mysqlType.startsWith(MysqlType_TINYINT)) {
            return StringUtils.isBlank(column.getValue()) ? 0 : column.getValue();
        } else if (mysqlType.startsWith(MysqlType_VARCHAR)) {
            return "'" + column.getValue() + "'";
        } else {
            return "'" + column.getValue() + "'";
        }
    }

}